/**
 * @file market_data_consumer.cpp
 * @brief 市场数据消费者类的实现
 * 
 * @details 该文件实现了市场数据消费者类，负责接收和处理来自交易所的市场数据更新。
 * 实现了丢包检测、快照同步和恢复机制，确保市场数据的完整性和连续性。
 * 
 * @author 原作者
 * @date 2023
 */

#include "market_data_consumer.h"

namespace Trading {
  /**
   * @brief 市场数据消费者类的构造函数实现
   * 
   * @details 初始化市场数据消费者对象，设置市场数据更新队列、日志记录器和多播套接字。
   * 在构造过程中，只初始化并连接到增量数据流的多播套接字，而不连接快照流。
   * 快照流只在需要恢复时才会连接。
   * 
   * @param client_id 客户端 ID，用于生成日志文件名
   * @param market_updates 市场数据更新的无锁队列指针
   * @param iface 网络接口名称
   * @param snapshot_ip 快照流的多播 IP 地址
   * @param snapshot_port 快照流的端口号
   * @param incremental_ip 增量流的多播 IP 地址
   * @param incremental_port 增量流的端口号
   */
  MarketDataConsumer::MarketDataConsumer(Common::ClientId client_id, Exchange::MEMarketUpdateLFQueue *market_updates,
                                         const std::string &iface,
                                         const std::string &snapshot_ip, int snapshot_port,
                                         const std::string &incremental_ip, int incremental_port)
      : incoming_md_updates_(market_updates), run_(false),
        logger_("trading_market_data_consumer_" + std::to_string(client_id) + ".log"),
        incremental_mcast_socket_(logger_), snapshot_mcast_socket_(logger_),
        iface_(iface), snapshot_ip_(snapshot_ip), snapshot_port_(snapshot_port) {
    // 创建接收回调函数，当有数据到达时调用 recvCallback 方法
    auto recv_callback = [this](auto socket) {
      recvCallback(socket);
    };

    // 设置增量流多播套接字的接收回调函数
    incremental_mcast_socket_.recv_callback_ = recv_callback;
    // 初始化增量流多播套接字，如果失败则触发断言
    ASSERT(incremental_mcast_socket_.init(incremental_ip, iface, incremental_port, /*is_listening*/ true) >= 0,
           "Unable to create incremental mcast socket. error:" + std::string(std::strerror(errno)));

    // 加入增量流多播组，如果失败则触发断言
    ASSERT(incremental_mcast_socket_.join(incremental_ip),
           "Join failed on:" + std::to_string(incremental_mcast_socket_.socket_fd_) + " error:" + std::string(std::strerror(errno)));

    // 设置快照流多播套接字的接收回调函数，但暂不初始化和连接
    snapshot_mcast_socket_.recv_callback_ = recv_callback;
  }

  /**
   * @brief 市场数据消费者线程的主循环
   * 
   * @details 该方法是市场数据消费者线程的主循环，负责从多播套接字读取并处理消息。
   * 实际的处理逻辑在 recvCallback() 和 checkSnapshotSync() 方法中实现。
   * 该方法持续调用多播套接字的 sendAndRecv() 方法，直到 run_ 标志被设置为 false。
   * 
   * @return void 无返回值
   */
  auto MarketDataConsumer::run() noexcept -> void {
    // 记录线程启动日志
    logger_.log("%:% %() %\n", __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_));
    
    // 主循环，当 run_ 为 true 时持续执行
    while (run_) {
      // 处理增量流多播套接字的数据
      incremental_mcast_socket_.sendAndRecv();
      // 处理快照流多播套接字的数据（如果已连接）
      snapshot_mcast_socket_.sendAndRecv();
    }
  }

  /**
   * @brief 通过订阅快照多播流开始快照同步过程
   * 
   * @details 该方法在检测到数据包丢失或系统刚启动时被调用，用于启动快照同步过程。
   * 它首先清除已排队的快照和增量消息，然后初始化快照多播套接字并订阅快照多播流。
   * 这使得系统可以接收完整的市场数据快照，从而恢复到当前状态。
   * 
   * @return void 无返回值
   */
  auto MarketDataConsumer::startSnapshotSync() -> void {
    // 清除已排队的快照和增量消息，准备重新同步
    snapshot_queued_msgs_.clear();
    incremental_queued_msgs_.clear();

    // 初始化快照多播套接字，如果失败则触发断言
    ASSERT(snapshot_mcast_socket_.init(snapshot_ip_, iface_, snapshot_port_, /*is_listening*/ true) >= 0,
           "Unable to create snapshot mcast socket. error:" + std::string(std::strerror(errno)));
    // 加入快照多播组（IGMP 多播订阅），如果失败则触发断言
    ASSERT(snapshot_mcast_socket_.join(snapshot_ip_), // IGMP multicast subscription.
           "Join failed on:" + std::to_string(snapshot_mcast_socket_.socket_fd_) + " error:" + std::string(std::strerror(errno)));
  }

  /**
   * @brief 检查是否可以从排队的市场数据更新中进行恢复/同步
   * 
   * @details 该方法检查是否可以从快照和增量市场数据流中排队的市场数据更新中进行恢复/同步。
   * 该方法的执行流程如下：
   * 1. 检查是否有快照消息，如果没有则返回
   * 2. 检查第一个快照消息是否为 SNAPSHOT_START 类型，如果不是则清除快照消息并返回
   * 3. 逐个检查快照消息，确保它们没有间隔并按序列号顺序排列
   * 4. 当遇到 SNAPSHOT_END 消息时，检查是否有完整的快照和快照之后的增量消息
   * 5. 如果恢复成功，将快照和增量消息按顺序发送到市场数据更新队列
   * 
   * @return void 无返回值
   */
  auto MarketDataConsumer::checkSnapshotSync() -> void {
    // 如果没有快照消息，则直接返回
    if (snapshot_queued_msgs_.empty()) {
      return;
    }

    // 获取第一个快照消息
    const auto &first_snapshot_msg = snapshot_queued_msgs_.begin()->second;
    // 检查第一个快照消息是否为 SNAPSHOT_START 类型
    if (first_snapshot_msg.type_ != Exchange::MarketUpdateType::SNAPSHOT_START) {
      logger_.log("%:% %() % Returning because have not seen a SNAPSHOT_START yet.\n",
                  __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_));
      // 如果不是，则清除快照消息并返回
      snapshot_queued_msgs_.clear();
      return;
    }

    // 用于存储最终要发送的市场数据更新事件
    std::vector<Exchange::MEMarketUpdate> final_events;

    // 标记是否有完整的快照
    auto have_complete_snapshot = true;
    size_t next_snapshot_seq = 0;
    // 遍历所有快照消息
    for (auto &snapshot_itr: snapshot_queued_msgs_) {
      logger_.log("%:% %() % % => %\n", __FILE__, __LINE__, __FUNCTION__,
                  Common::getCurrentTimeStr(&time_str_), snapshot_itr.first, snapshot_itr.second.toString());
      // 检查快照消息的序列号是否连续
      if (snapshot_itr.first != next_snapshot_seq) {
        have_complete_snapshot = false;
        logger_.log("%:% %() % Detected gap in snapshot stream expected:% found:% %.\n", __FILE__, __LINE__, __FUNCTION__,
                    Common::getCurrentTimeStr(&time_str_), next_snapshot_seq, snapshot_itr.first, snapshot_itr.second.toString());
        break;
      }

      // 将非开始和结束类型的快照消息添加到最终事件列表中
      if (snapshot_itr.second.type_ != Exchange::MarketUpdateType::SNAPSHOT_START &&
          snapshot_itr.second.type_ != Exchange::MarketUpdateType::SNAPSHOT_END)
        final_events.push_back(snapshot_itr.second);

      ++next_snapshot_seq;
    }

    if (!have_complete_snapshot) {
      logger_.log("%:% %() % Returning because found gaps in snapshot stream.\n",
                  __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_));
      snapshot_queued_msgs_.clear();
      return;
    }

    const auto &last_snapshot_msg = snapshot_queued_msgs_.rbegin()->second;
    if (last_snapshot_msg.type_ != Exchange::MarketUpdateType::SNAPSHOT_END) {
      logger_.log("%:% %() % Returning because have not seen a SNAPSHOT_END yet.\n",
                  __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_));
      return;
    }

    auto have_complete_incremental = true;
    size_t num_incrementals = 0;
    next_exp_inc_seq_num_ = last_snapshot_msg.order_id_ + 1;
    for (auto inc_itr = incremental_queued_msgs_.begin(); inc_itr != incremental_queued_msgs_.end(); ++inc_itr) {
      logger_.log("%:% %() % Checking next_exp:% vs. seq:% %.\n", __FILE__, __LINE__, __FUNCTION__,
                  Common::getCurrentTimeStr(&time_str_), next_exp_inc_seq_num_, inc_itr->first, inc_itr->second.toString());

      if (inc_itr->first < next_exp_inc_seq_num_)
        continue;

      if (inc_itr->first != next_exp_inc_seq_num_) {
        logger_.log("%:% %() % Detected gap in incremental stream expected:% found:% %.\n", __FILE__, __LINE__, __FUNCTION__,
                    Common::getCurrentTimeStr(&time_str_), next_exp_inc_seq_num_, inc_itr->first, inc_itr->second.toString());
        have_complete_incremental = false;
        break;
      }

      logger_.log("%:% %() % % => %\n", __FILE__, __LINE__, __FUNCTION__,
                  Common::getCurrentTimeStr(&time_str_), inc_itr->first, inc_itr->second.toString());

      if (inc_itr->second.type_ != Exchange::MarketUpdateType::SNAPSHOT_START &&
          inc_itr->second.type_ != Exchange::MarketUpdateType::SNAPSHOT_END)
        final_events.push_back(inc_itr->second);

      ++next_exp_inc_seq_num_;
      ++num_incrementals;
    }

    if (!have_complete_incremental) {
      logger_.log("%:% %() % Returning because have gaps in queued incrementals.\n",
                  __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_));
      snapshot_queued_msgs_.clear();
      return;
    }

    for (const auto &itr: final_events) {
      auto next_write = incoming_md_updates_->getNextToWriteTo();
      *next_write = itr;
      incoming_md_updates_->updateWriteIndex();
    }

    logger_.log("%:% %() % Recovered % snapshot and % incremental orders.\n", __FILE__, __LINE__, __FUNCTION__,
                Common::getCurrentTimeStr(&time_str_), snapshot_queued_msgs_.size() - 2, num_incrementals);

    snapshot_queued_msgs_.clear();
    incremental_queued_msgs_.clear();
    in_recovery_ = false;

    snapshot_mcast_socket_.leave(snapshot_ip_, snapshot_port_);;
  }

  /**
   * @brief 将消息排队到容器中
   * 
   * @details 该方法将市场数据更新消息排入 *_queued_msgs_ 容器中。
   * 根据第一个参数指定的消息来源（快照流或增量流），将消息存入相应的容器中。
   * 如果是快照消息，还会检查是否收到重复的序列号，如果是则清除快照消息容器。
   * 每次排队后都会调用 checkSnapshotSync() 方法检查是否可以进行快照同步。
   * 
   * @param is_snapshot 指示消息是否来自快照流的布尔值
   * @param request 市场数据更新消息指针
   */
  auto MarketDataConsumer::queueMessage(bool is_snapshot, const Exchange::MDPMarketUpdate *request) {
    if (is_snapshot) {
      // 如果是快照消息，检查是否收到重复的序列号
      if (snapshot_queued_msgs_.find(request->seq_num_) != snapshot_queued_msgs_.end()) {
        logger_.log("%:% %() % Packet drops on snapshot socket. Received for a 2nd time:%\n", __FILE__, __LINE__, __FUNCTION__,
                    Common::getCurrentTimeStr(&time_str_), request->toString());
        // 如果收到重复的序列号，清除快照消息容器
        snapshot_queued_msgs_.clear();
      }
      // 将快照消息按序列号存入快照消息容器
      snapshot_queued_msgs_[request->seq_num_] = request->me_market_update_;
    } else {
      // 将增量消息按序列号存入增量消息容器
      incremental_queued_msgs_[request->seq_num_] = request->me_market_update_;
    }

    // 记录当前快照和增量消息容器的大小以及消息详情
    logger_.log("%:% %() % size snapshot:% incremental:% % => %\n", __FILE__, __LINE__, __FUNCTION__,
                Common::getCurrentTimeStr(&time_str_), snapshot_queued_msgs_.size(), incremental_queued_msgs_.size(), request->seq_num_, request->toString());

    // 检查是否可以进行快照同步
    checkSnapshotSync();
  }

  /**
   * @brief 处理市场数据更新
   * 
   * @details 该方法处理从多播套接字接收到的市场数据更新。消费者需要使用套接字参数
   * 来确定该数据是来自快照流还是增量流。该方法的主要功能包括：
   * 1. 确定数据来源（快照流或增量流）
   * 2. 如果是快照消息但当前不在恢复模式，则丢弃该消息
   * 3. 如果检测到序列号间隔，则进入恢复模式
   * 4. 在恢复模式下，将消息排队并检查是否可以完成同步
   * 5. 如果不在恢复模式且收到的是按序的增量消息，则直接处理并发送到队列
   * 
   * @param socket 接收到数据的多播套接字指针
   * @return void 无返回值
   */
  auto MarketDataConsumer::recvCallback(McastSocket *socket) noexcept -> void {
    // 测量 UDP 读取时间
    TTT_MEASURE(T7_MarketDataConsumer_UDP_read, logger_);

    // 开始测量回调函数执行时间
    START_MEASURE(Trading_MarketDataConsumer_recvCallback);
    // 判断数据来源是快照流还是增量流
    const auto is_snapshot = (socket->socket_fd_ == snapshot_mcast_socket_.socket_fd_);
    // 如果是快照消息但当前不在恢复模式，则丢弃该消息
    if (UNLIKELY(is_snapshot && !in_recovery_)) { // 从快照流读取到市场数据更新但我们不在恢复模式，因此不需要它并丢弃它
      socket->next_rcv_valid_index_ = 0;

      logger_.log("%:% %() % WARN Not expecting snapshot messages.\n",
                  __FILE__, __LINE__, __FUNCTION__, Common::getCurrentTimeStr(&time_str_));

      return;
    }

    // 如果接收缓冲区中有足够的数据来处理市场数据更新
    if (socket->next_rcv_valid_index_ >= sizeof(Exchange::MDPMarketUpdate)) {
      size_t i = 0;
      // 遍历所有完整的市场数据更新消息
      for (; i + sizeof(Exchange::MDPMarketUpdate) <= socket->next_rcv_valid_index_; i += sizeof(Exchange::MDPMarketUpdate)) {
        // 将数据转换为市场数据更新消息
        auto request = reinterpret_cast<const Exchange::MDPMarketUpdate *>(socket->inbound_data_.data() + i);
        logger_.log("%:% %() % Received % socket len:% %\n", __FILE__, __LINE__, __FUNCTION__,
                    Common::getCurrentTimeStr(&time_str_),
                    (is_snapshot ? "snapshot" : "incremental"), sizeof(Exchange::MDPMarketUpdate), request->toString());

        // 记录当前是否已经在恢复模式
        const bool already_in_recovery = in_recovery_;
        // 如果已经在恢复模式或者检测到序列号间隔，则进入恢复模式
        in_recovery_ = (already_in_recovery || request->seq_num_ != next_exp_inc_seq_num_);

        // 如果在恢复模式下
        if (UNLIKELY(in_recovery_)) {
          // 如果刚刚进入恢复模式，则启动快照同步过程
          if (UNLIKELY(!already_in_recovery)) { // 如果我们刚刚进入恢复模式，通过订阅快照多播流开始快照同步过程
            logger_.log("%:% %() % Packet drops on % socket. SeqNum expected:% received:%\n", __FILE__, __LINE__, __FUNCTION__,
                        Common::getCurrentTimeStr(&time_str_), (is_snapshot ? "snapshot" : "incremental"), next_exp_inc_seq_num_, request->seq_num_);
            startSnapshotSync();
          }

          // 将市场数据更新消息排队，并检查是否可以成功完成快照恢复/同步
          queueMessage(is_snapshot, request);
        } else if (!is_snapshot) { // 不在恢复模式且收到的是按正确顺序且没有间隔的数据包，直接处理它
          logger_.log("%:% %() % %\n", __FILE__, __LINE__, __FUNCTION__,
                      Common::getCurrentTimeStr(&time_str_), request->toString());

          // 更新下一个预期的增量序列号
          ++next_exp_inc_seq_num_;

          // 将市场数据更新发送到队列
          auto next_write = incoming_md_updates_->getNextToWriteTo();
          *next_write = std::move(request->me_market_update_);
          incoming_md_updates_->updateWriteIndex();
          // 测量写入无锁队列的时间
          TTT_MEASURE(T8_MarketDataConsumer_LFQueue_write, logger_);
        }
      }
      // 移动未处理完的数据到缓冲区开头
      memcpy(socket->inbound_data_.data(), socket->inbound_data_.data() + i, socket->next_rcv_valid_index_ - i);
      socket->next_rcv_valid_index_ -= i;
    }
    // 结束测量回调函数执行时间
    END_MEASURE(Trading_MarketDataConsumer_recvCallback, logger_);
  }
}
